package com.tdy.cdc.app.dwd;

import com.tdy.cdc.app.BaseSqlApp;
import com.tdy.cdc.common.Constant;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.FlinkSourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author NanHuang
 * @Date 2023/3/13
 */
public class App06_DwdMhsTherapyRecord extends BaseSqlApp {
    public static void main(String[] args) {
        new App06_DwdMhsTherapyRecord().init(
                5006,
                "App6_DwdMhsTherapyRecord",
                2
        );
    }
    @Override
    protected void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 创建源表
        createSourceTables(tableEnv);
        // 2 join
        joinTables(tableEnv);
        // 3 写入Kafka
        writeToKafka(tableEnv);
    }

    private void writeToKafka(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table target (" +
                "RECORD_ID                          STRING," +
                "MED_ORG_NAME                       STRING," +
                "MED_ORG_CODE                       STRING," +
                "HOSPITAL_NAME                      STRING," +
                "HOSPITAL_CODE                      STRING," +
                "AGE_UNIT                           STRING," +
                "AGE                                STRING," +
                "SEX_NAME                           STRING," +
                "SEX_CODE                           STRING," +
                "ID_CARD_NUMBER                     STRING," +
                "PATIENT_VISIT_CATEGORY_CODE        STRING," +
                "PATIENT_VISIT_CATEGORY             STRING," +
                "INPATIENT_VISIT_FLOW_ID            STRING," +
                "INPATIENT_NO                       STRING," +
                "OUTPATIENT_VISIT_FLOW_ID           STRING," +
                "OUTPATIENT_NO                      STRING," +
                "PATIENT_ID                         STRING," +
                "PATIENT_NAME                       STRING," +
                "DOMAIN_ID                          STRING," +
                "BED_NAME                           STRING," +
                "VISIT_TIMES                        STRING," +
                "APPLY_SHEET_NO                     STRING," +
                "AUTHOR_CODE                        STRING," +
                "AUTHOR_NAME                        STRING," +
                "CUSTODIAN_CODE                     STRING," +
                "CUSTODIAN_NAME                     STRING," +
                "EXECUTOR_DATE                      STRING," +
                "ORDERS_EXEC_STAFF_ID               STRING," +
                "ORDERS_EXEC_STAFF_NAME             STRING," +
                "VISIT_DATE                         STRING," +
                "BED_NO                             STRING," +
                "DEPT_ID                            STRING," +
                "DEPT_NAME                          STRING," +
                "WARD_ID                            STRING," +
                "WARD_NAME                          STRING," +
                "INVASIVE_TREAT_FLAG                STRING," +
                "ALLERGIC_HISTORY_FLAG              STRING," +
                "ALLERGIC_HISTORY                   STRING," +
                "PATIENT_WEIGHT                     STRING," +
                "DISPOSE_INSTRUCTION                STRING," +
                "ORDERS_USE_COMMENTS                STRING," +
                "NEXT_TREAT_SCHEME                  STRING," +
                "FOLLOW_UP_DATE                     STRING," +
                "FOLLOW_UP_WAY_CODE                 STRING," +
                "FOLLOW_UP_WAY_NAME                 STRING," +
                "FOLLOW_UP_CYCLE_CODE               STRING," +
                "FOLLOW_UP_CYCLE_NAME               STRING," +
                "ROOM_NO                            STRING," +
                "ROOM_NAME                          STRING," +
                "DIAG_CODE                          STRING," +
                "DIAG_NAME                          STRING," +
                "VISIT_ID                           STRING," +
                "primary key (RECORD_ID) not enforced" +
                ")" +
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_THERAPY_RECORD));
        tableEnv.executeSql("insert into target select * from join_result");
    }

    private void joinTables(StreamTableEnvironment tableEnv) {
        // 1 设置数据的TTL（暂时30min）
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60));
        // 2 join操作
        Table joinResult = tableEnv.sqlQuery("select " +
                "s.unique_id as RECORD_ID , " +
                "hospital_name as MED_ORG_NAME , " +
                "hospital_code as MED_ORG_CODE , " +
                "hospital_name as HOSPITAL_NAME , " +
                "hospital_code as HOSPITAL_CODE , " +
                "age_unit as AGE_UNIT , " +
                "age as AGE , " +
                "sex as SEX_NAME , " +
                "sex_code as SEX_CODE , " +
                "identify_no as ID_CARD_NUMBER , " +
                "patient_typecode as PATIENT_VISIT_CATEGORY_CODE , " +
                "patient_type as PATIENT_VISIT_CATEGORY , " +
                "case when patient_typecode = '3' then visit_id else '' end as INPATIENT_VISIT_FLOW_ID , " +
                "inpatient as INPATIENT_NO , " +
                "case when patient_typecode <> '3' then visit_id else '' end as OUTPATIENT_VISIT_FLOW_ID , " +
                "out_patient_id as OUTPATIENT_NO , " +
                "patient_id as PATIENT_ID , " +
                "patient_name as PATIENT_NAME , " +
                "s.domain_id as DOMAIN_ID , " +
                "bed_name as BED_NAME , " +
                "visit_times as VISIT_TIMES , " +
                "appy_id as APPLY_SHEET_NO , " +
                "author_code as AUTHOR_CODE , " +
                "author_name as AUTHOR_NAME , " +
                "custodian_code as CUSTODIAN_CODE , " +
                "custodian_name as CUSTODIAN_NAME , " +
                "executor_date as EXECUTOR_DATE , " +
                "executor_code as ORDERS_EXEC_STAFF_ID , " +
                "executor_order as ORDERS_EXEC_STAFF_NAME , " +
                "visit_date as VISIT_DATE , " +
                "bed_no as BED_NO , " +
                "dept_code as DEPT_ID , " +
                "dept_name as DEPT_NAME , " +
                "wards_id as WARD_ID , " +
                "wards_name as WARD_NAME , " +
                "invasive_sign as INVASIVE_TREAT_FLAG , " +
                "allergy_history_mark as ALLERGIC_HISTORY_FLAG , " +
                "allergy_history as ALLERGIC_HISTORY , " +
                "weight as PATIENT_WEIGHT , " +
                "handling_guidance as DISPOSE_INSTRUCTION , " +
                "order_use_remark as ORDERS_USE_COMMENTS , " +
                "treatment_plan as NEXT_TREAT_SCHEME , " +
                "followup_date as FOLLOW_UP_DATE , " +
                "fu_way_code as FOLLOW_UP_WAY_CODE , " +
                "fu_way as FOLLOW_UP_WAY_NAME , " +
                "followup_recommendcode as FOLLOW_UP_CYCLE_CODE , " +
                "followup_recommend as FOLLOW_UP_CYCLE_NAME , " +
                "ward_id as ROOM_NO , " +
                "ward_name as ROOM_NAME , " +
                "diag_code as DIAG_CODE , " +
                "diag_name as DIAG_NAME ," +
                "visit_id as VISIT_ID " +
                "from hdsd00_06_03 s    " +
                "left join tdy_list_diag d on s.unique_id = d.unique_id and s.xds_type = d.xds_type    ");
        tableEnv.createTemporaryView("join_result",joinResult);
    }

    private void createSourceTables(StreamTableEnvironment tableEnv) {
        // hdsd00_06_03
        tableEnv.executeSql("create table hdsd00_06_03 (" +
                "pk                               string," +
                "upload_time                      string," +
                "status                           string," +
                "empi                             string," +
                "encounter_id                     string," +
                "bed_name                         string," +
                "ward_name                        string," +
                "wards_id                         string," +
                "patient_id                       string," +
                "patient_domain                   string," +
                "visit_domain                     string," +
                "visit_id                         string," +
                "visit_times                      string," +
                "unique_id                        string," +
                "xds_type                         string," +
                "xds_name                         string," +
                "effective_time                   string," +
                "xds_version                      string," +
                "domain_id                        string," +
                "patient_type                     string," +
                "out_patient_id                   string," +
                "patient_typecode                 string," +
                "inpatient                        string," +
                "appy_id                          string," +
                "identify_no                      string," +
                "patient_name                     string," +
                "sex_code                         string," +
                "sex                              string," +
                "age                              string," +
                "age_unit                         string," +
                "write_time                       string," +
                "author_code                      string," +
                "author_name                      string," +
                "custodian_code                   string," +
                "custodian_name                   string," +
                "executor_date                    string," +
                "executor_code                    string," +
                "executor_order                   string," +
                "visit_date                       string," +
                "bed_no                           string," +
                "ward_id                          string," +
                "dept_code                        string," +
                "dept_name                        string," +
                "wards_name                       string," +
                "hospital_code                    string," +
                "hospital_name                    string," +
                "invasive_sign                    string," +
                "allergy_history_mark             string," +
                "allergy_history                  string," +
                "weight                           string," +
                "handling_guidance                string," +
                "order_use_remark                 string," +
                "treatment_plan                   string," +
                "followup_date                    string," +
                "fu_way_code                      string," +
                "fu_way                           string," +
                "followup_recommendcode           string," +
                "followup_recommend               string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_06_03","App6_DwdMhsTherapyRecord"));
        // tdy_list_diag
        super.readTdyListDiag(tableEnv,"App6_DwdMhsTherapyRecord");
    }
}
